package cn.jly.boot.kafka.service.impl;

import cn.jly.boot.kafka.service.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Service;

/**
 * @PackageName cn.jly.boot.kafka.service.impl
 * @ClassName KafkaConsumerImpl
 * @Description
 * @Author 姬岚洋
 * @Date 2021/1/10 下午 9:27
 */
@Service
public class KafkaConsumerImpl implements KafkaConsumer {

    private static final String FIRST_TOPIC_NAME = "first";
    private static final String SECOND_TOPIC_NAME = "first";
    private static final String REDIRECT_TOPIC_NAME = "redirect_topic";
    /**
     * 简单监听
     * <p>
     * 如果我们想指定topic、指定partition、指定offset来消费呢？也很简单，@KafkaListener注解已全部为我们提供
     *
     * @param record
     * @KafkaListener(id = "consumer1", groupId = "jly-group", topicPartitions = {
     * @TopicPartition(topic = "first", partitions = {"0"}),
     * @TopicPartition(topic = "second", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
     * })
     * <p>
     * 属性解释：
     * <p>
     * ① id：消费者ID；
     * ② groupId：消费组ID；
     * ③ topics：监听的topic，可监听多个；
     * ④ topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
     * <p>
     * 上面onMessage监听的含义：
     * 监听first的0号分区，同时监听second的0号分区和second的1号分区里面offset从8开始的消息。
     * <p>
     * 注意：topics和topicPartitions不能同时使用；
     */
    @KafkaListener(
            topics = {FIRST_TOPIC_NAME, SECOND_TOPIC_NAME},
            errorHandler = "consumerAwareListenerErrorHandler",
            containerFactory = "concurrentKafkaListenerContainerFactory"
    )
    @Override
    public void onMessage(ConsumerRecord<?, ?> record) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.printf("简单消费--> topic: %s, 分区数：%s，值：%s %n", record.topic(), record.partition(), record.value());
    }

    /**
     * 消息转发
     * <p>
     * 在实际开发中，我们可能有这样的需求，应用A从TopicA获取到消息，经过处理后转发到TopicB，
     * 再由应用B监听处理消息，即一个应用处理完成后将该消息转发至其他应用，完成消息的转发。
     * <p>
     * 在SpringBoot集成Kafka实现消息的转发也很简单，只需要通过一个@SendTo注解，
     * 被注解方法的return值即转发的消息内容
     *
     * @param record
     * @return
     */
    @KafkaListener(topics = {REDIRECT_TOPIC_NAME})
    @SendTo(value = "first")
    public String redirectMessage(ConsumerRecord<?, ?> record) {
        return String.join("|", "redirect_message", record.value().toString());
    }
}
